package com.linys.scala.qf.day06_spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

/**
  * spark算子进阶练习
  */
object PartitionsRdd_02 {

  val it = (index: Int, iter: Iterator[(Int)]) => {
    iter.map(x => "[partID:" + index + ", value:" + x + "]")
  }

  def func[T](index: Int, iter: Iterator[(T)]): Iterator[String] = {
    iter.map(x => "[partID:" + index + " value:" + x + "]")
  }

  def main(args: Array[String]): Unit = {
    //遍历出集合中每一个元素
    val conf = new SparkConf().setAppName("OOP").setMaster("local")
    val sc = new SparkContext(conf)

    //遍历出集合中每一个元素
    /*
    mapPartitions是对每个分区中的数据进行迭代
    第一个参数是迭代器送对象, 第二个参数表示是否保留父RDD的partition分区信息
    第二个参数的值默认是false一般是不修改(父RDD就需要使用到宽窄依赖的问题)
    (f: Iterator[T] => Iterator[U],preservesPartitioning: Boolean = false)
    ps:第一个_是迭代器对象 第二个_是分区(集合)中数据
    如果RDD数据量不大,建议采用mapPartition算子代替map算子,可以加快数据量的处理数据
    但是如果RDD中数据量过大比如10亿条,不建议使用mapPartitions开进行数据遍历,可能出现oom内存溢出错误
    */
    val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 2)
    println(rdd1.partitions.size)
    val rdd2: RDD[Int] = rdd1.mapPartitions(_.map(_ * 10))
    println(rdd2.collect.toList)

    /*
    mapPartitionsWithIndex 是对rdd中每个分区的遍历出操作
    (f: (Int, Iterator[T]) => Iterator[U],preservesPartitioning: Boolean = false)
    参数是一个柯里化 第二个参数是一个隐式转换
    函数的作用和mapPartitions是类似,不过要提供两个采纳数,第一个参数是分区号
    第二个参数是分区中元素存储到Iterator中(就可以操作这个Iterator)
    第三个参数是否保留符RDD的partitoin
    */
    val rdd3: RDD[String] = rdd1.mapPartitionsWithIndex(it)
    println(rdd3.collect.toList)


    val rddd0 = rdd1.aggregate(0)(math.max(_, _), _ + _)
    println("rddd0=" + rddd0)
    val rddd3 = rdd1.aggregate(5)(math.max(_, _), _ + _)
    println("rddd3=" + rddd3)

    //排序
    /*
    sortBykey
    根据key进行排序,但是key必须实现Ordered接口
    可根据传入的true 或 false 决定是升序 还是 降序
    参数默认是true
     */
    val rdd4 = sc.parallelize(Array((3, "aa"), (6, "cc"), (2, "bb"), (1, "dd")))
    val sorted: RDD[(Int, String)] = rdd4.sortByKey()
    val sorted2: RDD[(Int, String)] = rdd4.sortBy(_._1, false)
    println(sorted.collect.toList)
    println(sorted2.collect.toList)
    /*
    sortBy与sortByKey类似,但是更灵活,可以用func先对数据进行处理,
    然后按处理后的数据比较结果排序
    第一个参数:是处理数据的方式
    第二个参数:是排序方式 默认是true 升序
     */
    val rdd5 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 10))
    val rdd5_1: RDD[Int] = rdd5.sortBy(x => x, true)
    println(rdd5_1.collect.toBuffer)

    //重新分区
    val rdd6 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8), 4)
    println("初始化分区:" + rdd6.partitions.size)
    /*
    更改分区repartition可以从少量分区改变为多分区因为会发生shuffle
    根据分区数，从新通过网络随机洗牌所有数据。
     */
    val reps1 = rdd6.repartition(2)
    println("通过repartiton调整分区后的分区:" + reps1.partitions.length)
    /*
    更改分区 不可以少量分区更改为多分区,因为不会发生shuffle
    缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。
     */
    val reps2 = rdd6.coalesce(5)
    println("通过coalesce调整分区后的分区:" + reps2.partitions.length)

    val rdd6_1 = sc.parallelize(List(("e", 5), ("c", 3), ("d", 4), ("c", 2), ("a", 1)), 2)
    //可以传入自定分区器, 也可以传入默认分区器 HashPartitioner
    val reps3: RDD[(String, Int)] = rdd6_1.partitionBy(new HashPartitioner(4))
    println(reps3.partitions.length)

    /*
    repartitionAndSortWithinPartitions算子是repartition算子的变种，与repartition算子不同的是，
    repartitionAndSortWithinPartitions在给定的partition内部进行排序，性能比repartition要高。
    ps: 必须是可排序的二元组 会根据key值排序
        参数可以是系统分区 也可以是自定义分区
   官方建议，如果需要在repartition重分区之后，还要进行排序，建议直接使用repartitionAndSortWithinPartitions算子。
   因为该算子可以一边进行重分区的shuffle操作，一边进行排序。shuffle与sort两个操作同时进行，
   比先shuffle再sort来说，性能可能是要高的
     */
    rdd4.repartitionAndSortWithinPartitions(new HashPartitioner(1)).foreach(println)

    //求和
    /*
    使用指定的reduce函数，将相同key的值聚合到一起
     */
    val rdd7 = sc.parallelize(Array(("tom", 1), ("jerry", 3), ("kitty", 2), ("jerry", 1), ("tom", 2), ("tom", 3), ("dog", 10)))
    val rdd7_1: RDD[(String, Int)] = rdd7.reduceByKey(_+_)
    println("-------------------------------------华丽的分割线1------------------------------------------")
    println(rdd7_1.collect.toList)
    /*
    aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,combOp: (U, U) => U)
    在kv对的RDD中，，按key将value进行分组合并，合并时，将每个value和初始值作为seq函数的参数，
    进行计算，返回的结果作为一个新的kv对，然后再将结果按照key进行合并，
    最后将每个分组的value传递给comb函数进行计算（先将前两个value进行计算，将返回结果和下一个value传给comb函数，以此类推），
    将key与计算结果作为一个新的kv对输出。 seqOp函数用于在每一个分区中用初始值逐步迭代value，combOp函数用于合并每个分区中的结果
    zeroValue是初始值(默认值) seqOP局部聚合(分区)  combOp 全局聚合
     */
    val rdd8 = sc.parallelize(List(("cat", 2), ("cat", 5), ("pig", 10), ("dog", 3), ("dog", 4), ("cat", 4)), 2)
    println("-------------------------------------华丽的分割线2------------------------------------------")
    println(rdd8.mapPartitionsWithIndex(func).collect.toList)
    /*
    因为第一次给的数值每个分区中是没有相同key所有都是最大值,所有就相当于值都值相加了
    第二次将同一个分区中的key有相同
    首先会根据相同key来进行计算,以cat为例先会和初始值-进行计算比较留下最大值
    然后会的等待第二分区完成计算,然后在进行一个全局的聚合
    */
    val value: RDD[(String, Int)] = rdd8.aggregateByKey(0)(math.max(_, _), _ + _)
    println("-------------------------------------华丽的分割线3------------------------------------------")
    println(value.collect.toBuffer)
    /*
    combineByKey[C](createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C)
    对相同Key，把Value合并成一个集合.
    createCombiner: combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，
                   要么就和之前的某个元素的键相同。如果这是一个新的元素,combineByKey()
                   会使用一个叫作 createCombiner() 的函数来创建那个键对应的累加器的初始值
    mergeValue: 如果这是一个在处理当前分区之前已经遇到的键， 它会使用 mergeValue()
                 方法将该键的累加器对应的当前值与这个新的值进行合并
    mergeCombiners: 由于每个分区都是独立处理的， 因此对于同一个键可以有多个累加器。
                    如果有两个或者更多的分区都有对应同一个键的累加器，
                    就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并。
     */
    val rdd9 = sc.parallelize(List(("cat", 2), ("cat", 5), ("pig", 10), ("dog", 3), ("dog", 4), ("cat", 4)), 2)
    val rdd9_1: RDD[(String, Int)] = rdd9.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
    println(rdd9_1.collect().toList)


  }

}
